package org.zn.note.java8;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;
import java.util.stream.LongStream;

/**
 * JDK 1.7 引入的 分支/合并框架
 * 分支/合并框架的目的是以递归方式将可以并行的任务拆分成更小的任务，然后将每个子任务的结果合并起来生成整体结果。
 * 它是 {@link ExecutorService} 接口的一个实现，它把子任务分配给线程池（称为ForkJoinPool）中的工作线程。
 * <p>
 * {@link ForkJoinPool} 一般是单例的，实例化一次就可以了
 * 一般流程：
 * 1、自己的Task需要实现 {@link RecursiveTask} extend {@link ForkJoinTask}，只需要实现一个 compute 方法
 * 2、调用 {@link ForkJoinPool#invoke(ForkJoinTask)} 传入自己的task
 * <p>
 * 最佳实践：
 * 1、{@link ForkJoinTask#join()} 会阻塞。在当前线程下：
 * ① leftTask.fork();
 * ② rightTask.compute(); // 因为在当前线程中，这里是顺序的，默认就是等right执行完
 * ③ leftTask.join(); // 等left执行完
 * ④ Reduce leftTaskResult + rightTaskResult // 都执行完了，归约
 * 2、不要在task内部调用 ForkJoinPool.invoke(ForkJoinTask)，task里一般调用fork、compute。顺序代码中才调用invoke(用于启动)
 * 3、fork可以把任务排进ForkJoinPool，好像直接 leftTask.fork; rightTask.fork 是很自然的，那为什么 rightTask.compute ？
 * 其实是可以：
 * left.fork
 * right.fork
 * left.join
 * right.join
 * Reduce left + right
 * 但没必要，因为当先线程也是闲着等待，不如复用，所以left用新线程，right用当前线程就可以了。
 * 最好的写法还是 1 中的那样。
 * <p>
 * 关于是否选择并行，在Stream并行里的那些要点，这里同样适用（因为Stream的并行就是用这个实现的）
 * <p>
 * 工作窃取(work stealing)
 * 按理说，我只有4个核，也就是默认只起4个线程，那么我的子任务就分成4份就好了呢？
 * 不是的，一般分出大量的小任务是更好的。
 * 实际上，各种原因，每个核每个线程有快有慢，
 * 刚开始的任务可能会“平均”分给每个线程，但是在执行的过程中，执行快的线程如果没有任务了，会去执行慢的线程那边“偷”个任务来执行。
 * 所以更多的小任务，更有利于工作线程之间的平衡负载。
 */
public class Ch07_ForkJoin {

    public static void main(String[] args) {
        long sum = forkJoinSum(10000000);
        System.out.println(sum);
    }

    // 前N数求和
    private static Long forkJoinSum(long n) {
        long[] numbers = LongStream.rangeClosed(1, n).toArray();
        ForkJoinTask<Long> task = new ForkJoinSumCalculator(numbers);
        return new ForkJoinPool().invoke(task);
    }

    private static class ForkJoinSumCalculator extends RecursiveTask<Long> {

        private final long[] numbers; // 需要求和的数据数组
        private final int start, end; // 子任务需要处理的数据，在整个数据数组中的开始结束位置
        public static final long THRESHOLD = 10_000; // 不再将任务分解为子任务的数组大小 （这种定义是jdk1.7引入的）

        public ForkJoinSumCalculator(long[] numbers) {
            this(numbers, 0, numbers.length);
        }

        // 私有构造方法
        private ForkJoinSumCalculator(long[] numbers, int start, int end) {
            this.numbers = numbers;
            this.start = start;
            this.end = end;
        }

        @Override
        protected Long compute() {
            System.out.printf("%s\n", Thread.currentThread().getName());
            int length = end - start;
            if (length <= THRESHOLD) { // 不可再分的任务，直接顺序计算求和
                return computeSequentially();
            }
            ForkJoinSumCalculator leftTask =
                    new ForkJoinSumCalculator(numbers, start, start + length / 2);
            leftTask.fork();
            ForkJoinSumCalculator rightTask =
                    new ForkJoinSumCalculator(numbers, start + length / 2, end);
            Long rightResult = rightTask.compute();
            Long leftResult = leftTask.join();
            return leftResult + rightResult;
        }

        // 不可再分的任务，开始顺序计算
        private long computeSequentially() {
            long sum = 0;
            for (int i = start; i < end; i++) {
                sum += numbers[i];
            }
            return sum;
        }
    }
}